package com.wuwangfu.window.event;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @Author jcshen
 * @Date 2023-02-24
 * @PackageName:com.wuwangfu.window.event
 * @ClassName: EventTimeTumbleWindowB
 * @Description:
 * @Version 1.0.0
 *
 * 先keyBy，再按照EventTime划分滚动窗口
 *
 * SocketSource的并行度为 1，不在source对应的DataStream生成watermark，而是先调用map生成新的DataStream，
 * 在新的DataStream上生成watermark
 *
 */
public class EventTimeTumbleWindowB {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);
        /**
         * 1000,spark,1
         * 2000,hive,1
         * 4999,spark,1
         * 4999,spark,2
         *
         * 6666,hive,3
         * 7777,flink,5
         *---------------
         * 触发窗口（下一个窗口），只跟生成watermark的DataStream的分区有关，
         * 单分区情况下，最大时间（EventTime） - 乱序延迟时间 >= 窗口的结束时间，就会触发窗口
         * 多分区情况下，每个分区里面的 最大时间（EventTime） - 乱序延迟时间 （都要） >= 窗口的结束时间，才会触发
         *
         * 10000,spark,1
         * 12000,flink,6
         */
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> maped = line.map(new MapFunction<String, Tuple3<Long, String, Integer>>() {
            @Override
            public Tuple3<Long, String, Integer> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple3.of(Long.parseLong(fields[0]), fields[1], Integer.parseInt(fields[2]));
            }
        }).setParallelism(2);
        //生成watermark
        SingleOutputStreamOperator<Tuple3<Long, String, Integer>> dataWithWatermark = maped.assignTimestampsAndWatermarks(
                new BoundedOutOfOrdernessTimestampExtractor<Tuple3<Long, String, Integer>>(Time.seconds(0)) {
                    @Override
                    public long extractTimestamp(Tuple3<Long, String, Integer> element) {
                        return element.f0;
                    }
                });
        //映射，分组
        KeyedStream<Tuple, Tuple> keyed = dataWithWatermark.project(1, 2).keyBy(0);
        //划分窗口
        WindowedStream<Tuple, Tuple, TimeWindow> windowed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));
        //聚合
        windowed.sum(1).print();

        env.execute();
    }
}
